package experiment2;

import java.io.FileInputStream;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FSDataOutputStream;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

public class HDFSDemo {

    /**

     * 判断路径是否存在

     */

    public static boolean test(Configuration conf, String path) {

        try (FileSystem fs = FileSystem.get(conf)) {

            return fs.exists(new Path(path));

        } catch (IOException e) {

            e.printStackTrace();

            return false;

        }

    }

    /**

     * 复制文件到指定路径 若路径已存在，则进行覆盖

     */

    public static void HDFSDemo(Configuration conf,

                                         String localFilePath, String remoteFilePath) {

        Path localPath = new Path(localFilePath);

        Path remotePath = new Path(remoteFilePath);

        try (FileSystem fs = FileSystem.get(conf)) {

/* fs.HDFSDemo
第一个参数表示是否删除源文件，第二个参数表示是否覆盖 */

            fs.HDFSDemo(false, true, localPath, remotePath);

        } catch (IOException e) {

            e.printStackTrace();

        }

    }

    /**

     * 追加文件内容

     */

    public static void appendToFile(Configuration conf, String
            localFilePath,

                                    String remoteFilePath) {

        Path remotePath = new Path(remoteFilePath);

        try (FileSystem fs = FileSystem.get(conf);

             FileInputStream in = new FileInputStream(localFilePath) {

            FSDataOutputStream out = fs.append(remotePath);

            byte[] data = new byte[1024];

            int read = -1;

            while ((read = in.read(data)) > 0) {

                out.write(data, 0, read);

            }

            out.close();

        } catch (IOException e) {

            e.printStackTrace();

        }

    }

    /**

     * 主函数

     */

    public static void main(String[] args) {

        Configuration conf = new Configuration();

        conf.set("fs.defaultFS", "hdfs://node1:9000");

        String localFilePath = ""; // 本地路径

        String remoteFilePath = "test/test.txt"; // HDFS路径

        String choice = "append"; // 若文件存在则追加到文件末尾

// String choice = “overwrite”; // 若文件存在则覆盖

        try {

            /* 判断文件是否存在 */

            boolean fileExists = false;

            if (HDFSDemo.test(conf, remoteFilePath)) {

                fileExists = true;

                System.out.println(remoteFilePath + " 已存在.");

            } else {

                System.out.println(remoteFilePath + " 不存在.");

            }

            /* 进行处理 */

            if (!fileExists) { // 文件不存在，则上传

                HDFSDemo.HDFSDemo(conf, localFilePath,

                        remoteFilePath);

                System.out.println(localFilePath + " 已上传至 " + remoteFilePath);

            } else if (choice.equals("overwrite")) { // 选择覆盖

                HDFSDemo.HDFSDemo(conf, localFilePath,

                        remoteFilePath);

                System.out.println(localFilePath + " 已覆盖 " + remoteFilePath);

            } else if (choice.equals("append")) { // 选择追加

                HDFSDemo.appendToFile(conf, localFilePath,

                        remoteFilePath);

                System.out.println(localFilePath + " 已追加至 " + remoteFilePath);

            }

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

}
